package com.bangcommunity.bbframe.mq.kafka.interceptor;

import com.bangcommunity.bbframe.common.annotation.BeanOrder;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.bangcommunity.bbframe.mq.IQtMessager;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

/**
 * Created by tanghc on 17/2/4.
 */
@BeanOrder(order = -1)
public class QtMonitorKafkaConsumerInterceptor implements ConsumerInterceptor<IQtMessager, IQtMessager> {

    @Override
    public ConsumerRecords<IQtMessager, IQtMessager> onConsume(ConsumerRecords<IQtMessager, IQtMessager> records) {
        if (null != records) {
            for (ConsumerRecord<IQtMessager, IQtMessager> record : records) {
                IQtMessager msg = record.value();
                if (null != msg) {
                    System.out.println("=====test monitor kafka===" + StringUtils.join(msg.getAttach().entrySet()));
                }
            }
        }
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
